package xin.marcher.module.message.mq.consumer.listener;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import xin.marcher.framework.common.util.json.JsonUtils;
import xin.marcher.module.common.constants.CacheConstant;
import xin.marcher.module.common.constants.DateConstant;
import xin.marcher.module.common.redis.RedisCache;
import xin.marcher.module.common.utils.JsonUtil;
import xin.marcher.module.message.cache.CacheQueue;
import xin.marcher.module.message.domain.bo.MessageCache;

import javax.annotation.Resource;
import java.util.Date;
import java.util.List;
import java.util.concurrent.BlockingQueue;

/**
 * 缓存消息处理
 */
@Slf4j
@Component
public class DataMessageCacheListener implements MessageListenerConcurrently {

    @Resource
    private CacheQueue cacheQueue;

    @Resource
    private RedisCache redisCache;

    /**
     * 上次记录消费的时间
     * TODO 多实例下该时间未同步, 会导致多个实例重复消费
     */
    private long lastTimestamp = -1L;

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {
        for (MessageExt messageExt : list) {
            try {
                String messageData = new String(messageExt.getBody());
                log.info("DataMessageCacheListener缓存数据变更刷新，消息内容：{}", messageData);
                MessageCache messageCache = JsonUtils.parseObject(messageData, MessageCache.class);

                // 根据消息的缓存 key，获取到对应的内存队列，分散队列提高处理效率，并保证单 key 的执行不会并发
                BlockingQueue blockingQueue = cacheQueue.getBlockingQueue(messageCache.getCacheKey());
                // 入队
                blockingQueue.offer(messageCache);

                // 记录最新的消费数据时间
                setCacheRefreshTime();
            } catch (Exception e) {
                log.error("consume error, 缓存消息写入队列失败", e);
                // 本次消费失败，下次重新消费
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }

    /**
     * 记录最新的消费时间，为避免无效的set操作，这里控制每秒最多一次set
     */
    private synchronized void setCacheRefreshTime() {
        // 获取到当前的时间，精确到秒
        long timestamp = System.currentTimeMillis() / 1000;

        // 同一个时间则默认不处理
        if (lastTimestamp == timestamp) {
            return;
        }
        // 标记最新的MQ消息接收时间(如果实际落DB的数据时间和消费的数据时间间隔时间超过1分钟则开始同步DB的数据兜底缓存刷新)
        redisCache.set(CacheConstant.CACHE_ROCKET_TIME_KEY, DateFormatUtils.format(new Date(), DateConstant.DATE_TIME_FORMAT_PATTERN), 0);
        lastTimestamp = timestamp;
    }
}
